package com.lxr.iot;

import com.lxr.iot.auto.MqttListener;
import com.lxr.iot.auto.MqttMessageListener;
import com.lxr.iot.bootstrap.MqttProducer;
import com.lxr.iot.bootstrap.Producer;
import com.lxr.iot.properties.ConnectOptions;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * @author lxr
 * @create 2018-01-12 15:14
 **/
@Slf4j
@Service
@MqttMessageListener(qos = MqttQoS.AT_LEAST_ONCE, topic = "/topic/test/one")
public class SubListener implements MqttListener {

    @Override
    public void callBack(String topic, String msg) {
        log.info("topic : {} , msg : {}", topic, msg);
        // TODO 收到消息后,发布消息.
        this.pub("/callback/ok","ok" , MqttQoS.EXACTLY_ONCE);
    }

    @Override
    public void callThrowable(Throwable e) {
        log.error("exception", e);
    }

    public void pub(String topic,String message,MqttQoS mqttQoS) {
        Producer producer = null;
        try {
            producer = new MqttProducer();
            ConnectOptions connectOptions = new ConnectOptions();
            connectOptions.setBacklog(1024);
            connectOptions.setConnectTime(1000l);
            connectOptions.setSsl(false);
            connectOptions.setServerIp("192.168.1.10");
            connectOptions.setPort(1883);
            connectOptions.setBossThread(1);
            connectOptions.setWorkThread(8);
            connectOptions.setMinPeriod(10);
            connectOptions.setRevbuf(1024);
            connectOptions.setSndbuf(1024);
            connectOptions.setTcpNodelay(true);
            connectOptions.setKeepalive(true);
            ConnectOptions.MqttOpntions mqttOpntions = new ConnectOptions.MqttOpntions();
            mqttOpntions.setClientIdentifier("111");
            mqttOpntions.setHasPassword(false);
            mqttOpntions.setHasPassword(false);
            connectOptions.setMqtt(mqttOpntions);
            producer.setMqttListener(new MqttListener() {
                @Override
                public void callBack(String topic, String msg) {
                    System.out.print("========================================"+topic+msg);
                }
                @Override
                public void callThrowable(Throwable e) {
                    log.error("" , e);
                }
            });
            producer.connect(connectOptions);
//        producer.sub(SubMessage.builder().qos(MqttQoS.AT_LEAST_ONCE).topic("/t1/t2").build());
            producer.pub(topic ,message,mqttQoS.value());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}
